#include <assert.h>
#include <google/protobuf/message.h>
#include <google/protobuf/text_format.h>

#include "echo.pb.h"
#include "zrpc/base/buffer.h"
#include "zrpc/base/logger.h"
#include "zrpc/net/event_loop.h"
#include "zrpc/net/tcp_connection.h"
#include "zrpc/net/tcp_server.h"
#include "zrpc/rpc/buffer_stream.h"

using namespace zrpc;

class EchoServer {
 public:
  EchoServer(std::string ip, int port) : server_(&loop_, ip, port, nullptr) {}

  void Start() {
    server_.SetThreadNum(0);
    // 设置连接回调注册监听
    server_.SetConnectionCallback(std::bind(&EchoServer::ConnectionCallback,
                                            this, std::placeholders::_1));
    // 设置消息回调注册监听
    // server_.SetMessageCallback(std::bind(&EchoServer::MessageCallback, this,
    //                                     std::placeholders::_1,
    //                                     std::placeholders::_2));
    // server_.SetMessage1Callback(std::bind(&EchoServer::Message1Callback,
    // this,
    //                                      std::placeholders::_1,
    //                                      std::placeholders::_2));

    /*  server_.SetMessage1Callback(std::bind(&EchoServer::Message1Callback,
         this, std::placeholders::_1, std::placeholders::_2));*/
    server_.Start();
  }

  void ConnectionCallback(const std::shared_ptr<TcpConnection>& conn) {
    if (conn->Connected()) {
      assert(socket::SetKeepAlive(conn->GetSockfd(), 1));
      LOG_INFO << "connect";
    } else {
      LOG_INFO << "disconnect";
    }
  }

  void Loop() { loop_.Run(); }

  // ParseFromArray
  void MessageCallback_ParseFromArray(
      const std::shared_ptr<TcpConnection>& conn, Buffer* buf) {
    while (buf->ReadableBytes() >= sizeof(uint32_t)) {
      int16_t data_bytes = buf->PeekInt16();
      if (buf->ReadableBytes() >= data_bytes) {
        buf->RetrieveInt32();
        echo::PingRequest request;
        assert(request.ParseFromArray(buf->Peek(), data_bytes - kMsgHeadSize));
        buf->Retrieve(data_bytes - kMsgHeadSize);
        {
          Buffer reply_buf;
          echo::PongRepsonse response;
          response.set_id(message_);
          std::string str;
          assert(response.SerializeToString(&str));
          reply_buf.Append(str);
          reply_buf.PrependInt16(0);
          reply_buf.PrependInt16(reply_buf.ReadableBytes() + sizeof(int16_t));
          conn->Send(&reply_buf);
        }
      } else {
        break;
      }
    }
  }

  // ZeroCopyInputStream
  void MessageCallback_ZeroCopyInputStream(
      const std::shared_ptr<TcpConnection>& conn, Buffer* buf) {
    while (buf->ReadableBytes() >= sizeof(uint32_t)) {
      int16_t data_bytes = buf->PeekInt16();
      if (buf->ReadableBytes() >= data_bytes) {
        buf->RetrieveInt32();
        BufferInputStream input(buf, data_bytes - kMsgHeadSize);
        echo::PingRequest request;
        request.ParseFromZeroCopyStream(&input);
        buf->Retrieve(data_bytes - kMsgHeadSize);
        {
          Buffer reply_buf;
          echo::PongRepsonse response;
          response.set_id(message_);
          std::string str;
          assert(response.SerializeToString(&str));
          reply_buf.Append(str);
          reply_buf.PrependInt16(0);
          reply_buf.PrependInt16(reply_buf.ReadableBytes() + sizeof(int16_t));
          conn->Send(&reply_buf);
        }
      } else {
        break;
      }
    }
  }

  const static uint32_t kMsgHeadSize = static_cast<uint32_t>(sizeof(Message));

  const char* GetMessageBuf(const zrpc::Message& msg) {
    if (msg.size < kMsgHeadSize) {
      return nullptr;
    }

    return (reinterpret_cast<const char*>(&msg) + kMsgHeadSize);
  }

  void Message1Callback(const std::shared_ptr<TcpConnection>& conn,
                        const zrpc::Message& msg) {
    const char* buf = GetMessageBuf(msg);
    if (nullptr == buf) {
      return;
    }

    if (msg.size > kMsgHeadSize) {
      echo::PingRequest request;
      assert(request.ParseFromArray(buf, msg.size - kMsgHeadSize));

      {
        Buffer buf;
        echo::PongRepsonse response;
        response.set_id(message_);
        std::string str;
        assert(response.SerializeToString(&str));
        buf.Append(str);
        buf.PrependInt16(0);
        buf.PrependInt16(buf.ReadableBytes() + sizeof(int16_t));
        conn->Send(&buf);
      }
    }
  }

  void TimerCallback() { LOG_INFO << "repeat timer callback"; }

 private:
  EventLoop loop_;
  TcpServer server_;
  std::string message_;
};

int main() {
  EchoServer server("10.23.7.160", 6379);
  server.Start();
  server.Loop();
}
